package com.ada.spark.rddoperator

import org.apache.spark.{SparkConf, SparkContext}

/**
  * 作用：在kv对的RDD中，，按key将value进行分组合并，合并时，将每个value和初始值作为seq函数的参数，进行计算，返回的结果作为一个新的kv对，然后再将结果按照key进行合并，最后将每个分组的value传递给combine函数进行计算（先将前两个value进行计算，将返回结果和下一个value传给combine函数，以此类推），将key与计算结果作为一个新的kv对输出。
  */
object Spark22_aggregateByKey {

    def main(args: Array[String]): Unit = {
        //创建SparkConf
        val conf = new SparkConf().setAppName("Spark22_aggregateByKey").setMaster("local[*]")
        //创建Spark上下文对象
        val sc = new SparkContext(conf)

        val rdd = sc.parallelize(List(("a", 3), ("a", 2), ("c", 4), ("b", 3), ("c", 6), ("c", 8)), 2)

        //分区情况
        rdd.glom().collect().foreach(datas => println(datas.mkString(",")))
        //分区0：(a,3),(a,2),(c,4)
        //分区1：(b,3),(c,6),(c,8)

        //取出每个分区相同key对应值的最大值，然后相加
        //参数：(zeroValue:U,[partitioner: Partitioner]) (seqOp: (U, V) => U,combOp: (U, U) => U)
        val agg = rdd.aggregateByKey(0)(math.max(_, _), _ + _)

        println(agg.collect().mkString(","))
        //(b,3),(a,3),(c,12)
    }

}
